package io.github.ali.commons.lock.impl;

import org.apache.commons.lang3.BooleanUtils;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import io.github.ali.commons.util.Asserts;

/**
 * 基于Redis实现的分布式锁
 */
public class RedisLocker extends AbstractLocker {
    //通过Lua脚本删除key,确保原子操作
    private static final String DEL_LUA_SCRIPT = "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) else return 0 end";
    //通过Lua脚本setnx+expire,确保原子操作
    private static final String SETNX_LUA_SCRIPT = "if redis.call('SETNX', KEYS[1], ARGV[1]) == 1 then return redis.call('EXPIRE', KEYS[1], ARGV[2]) else return 0 end";

    private final RedisTemplate<String, String> redisTemplate;
    private final RedisSerializer<String> keySerializer;
    private final RedisSerializer<String> valueSerializer;

    public RedisLocker(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.keySerializer = new StringRedisSerializer();
        this.valueSerializer = new StringRedisSerializer();
    }

    @Override
    protected void doLock(String key, String value, long expiring) {
        //此方案用于spring-data-redis 2.0版本
        // Boolean result = redisTemplate.execute((RedisCallback<Boolean>) redisConnection ->
        //         redisConnection.set(serializeKey(key), serializeValue(value), Expiration.seconds(expiring), RedisStringCommands.SetOption.SET_IF_ABSENT));
        Boolean result = redisTemplate.execute((RedisCallback<Boolean>) redisConnection ->{
        	byte[] keyBytes=keySerializer.serialize(key);
        	byte[] valueBytes=valueSerializer.serialize(value);
        	if(expiring>0){
        		return redisConnection.eval(SETNX_LUA_SCRIPT.getBytes(), ReturnType.BOOLEAN, 1, keyBytes, valueBytes, String.valueOf(expiring).getBytes());
        	}
        	return redisConnection.setNX(keyBytes, valueBytes);
        });
        Asserts.check(BooleanUtils.toBoolean(result), "redis setNX() returned false");
    }

    @Override
    protected void doUnlock(String key, String value) {
        redisTemplate.execute((RedisCallback<Boolean>) redisConnection ->
                redisConnection.eval(DEL_LUA_SCRIPT.getBytes(), ReturnType.BOOLEAN, 1, keySerializer.serialize(key), valueSerializer.serialize(value)));
    }
}
